package com.ts.blog.batch.mysql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


/**
 * @classname:
 * @description:
 * @author: yishiyong
 * @create: 2018-12-25
 */
public class SparkMysql {
    public static org.apache.log4j.Logger logger = org.apache.log4j.Logger.getLogger(SparkMysql.class);

    public static void main(String[] args) {
        JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf().setAppName("SparkMysql").setMaster("local[5]"));
        SQLContext sqlContext = new SQLContext(sparkContext);
        //写入的数据内容
        List lists = new ArrayList();
        for (int i=1;i<100;i++){
            StringBuffer buf = new StringBuffer();
            buf.append("java").append(i).append(" spring").append(i).append(" ").append(i).append(" 2019-01-01 ").append(i);
            lists.add(buf.toString());
            //Arrays.asList("java spring 12 2018-01-01 12","java1 spring1 12 2018-01-01 12","java2 spring2 12 2018-01-01 12")
        }

        JavaRDD<String> personData = sparkContext.parallelize(lists);
        //数据库内容
        String url = "jdbc:mysql://172.18.101.97:3306/lnaudit";
        //查找的表名
        String table = "blog";
        //增加数据库的用户名(user)密码(password),指定test数据库的驱动(driver)
        Properties connectionProperties = new Properties();
        connectionProperties.put("user","lnaudit");
        connectionProperties.put("password","lnaudit");
        connectionProperties.put("driver","com.mysql.jdbc.Driver");

        /**
         * 第一步：在RDD的基础上创建类型为Row的RDD
         */
        //将RDD变成以Row为类型的RDD。Row可以简单理解为Table的一行数据
        JavaRDD<Row> personsRDD = personData.map(new Function<String,Row>(){
            public Row call(String line) throws Exception {
                String[] splited = line.split(" ");
                //new SimpleDateFormat("yyyy-MM-dd").parse(splited[3])
                return RowFactory.create(splited[0],splited[1],Long.valueOf(splited[2])
                        ,java.sql.Date.valueOf(splited[3]),Long.valueOf(splited[4]));
            }
        });

        /**
         * 第二步：动态构造DataFrame的元数据。
         */
        List structFields = new ArrayList();
        structFields.add(DataTypes.createStructField("topic",DataTypes.StringType,true));
        structFields.add(DataTypes.createStructField("subtopic",DataTypes.StringType,true));
        structFields.add(DataTypes.createStructField("likes",DataTypes.LongType,true));
        structFields.add(DataTypes.createStructField("time",DataTypes.DateType,true));
        structFields.add(DataTypes.createStructField("visits",DataTypes.LongType,true));

        //构建StructType，用于最后DataFrame元数据的描述
        StructType structType = DataTypes.createStructType(structFields);

        /**
         * 第三步：基于已有的元数据以及RDD<Row>来构造DataFrame
         */
        Dataset<Row> personsDF = sqlContext.createDataFrame(personsRDD,structType);

        /**
         * 第四步：将数据写入到person表中
         */
        personsDF.write().mode("append").jdbc(url,table,connectionProperties);

        //停止SparkContext
        sparkContext.stop();
    }
}